package com.apex.spark;

import com.alibaba.fastjson.JSONObject;
import com.apex.env.BaseEnv;
import com.apex.spark.utils.ConfigKeyName;
import com.apex.spark.utils.SchemaUtil;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigObject;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.Seconds;
import org.apache.spark.streaming.StreamingContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * 初始化spark执行环境,所有插件环境的入口
 */
public class SparkEnvironment implements BaseEnv {

    private SparkSession sparkSession;

    private SparkConf sparkConf;

    private StreamingContext streamingContext;

    private Config config;

    private boolean isHiveTable;

    private StructType schema;

    private String tableName;

    private List<String> sqlAll;

    /**
     *
     * @param mode 用于判断执行模式时流模式还是批处理模式
     */
    @Override
    public void prepare(Boolean mode) {
        isHiveTable = config.getBoolean(ConfigKeyName.SPARK_RUN_HIVE_MODE);
        createSparkSession();
//        createStreamingContext();
    }

    /**
     * 获取配置文件管理工具
     * @return config
     */
    @Override
    public Config getConfig() {
        return config;
    }

    /**
     * 设置配置环境管理工具执行环境
     * @param config config
     */
    @Override
    public void setConfig(Config config) {
        this.config = config;
    }

    /**
     * 创建一个sparkSession执行环境,并判断是否启用hive模式
     */
    private void createSparkSession(){
        setSparkConf();
        if (isHiveTable){
            sparkSession = SparkSession
                    .builder()
                    .config(sparkConf)
                    .enableHiveSupport()
                    .getOrCreate();
        }else {
            sparkSession = SparkSession
                    .builder()
                    .config(sparkConf)
                    .getOrCreate();
        }

    }

    /**
     * 配置spark环境执行参数
     */
    private void setSparkConf(){
        sparkConf = new SparkConf();
        sparkConf.set("spark.io.compression.codec","snappy");
        ConfigObject configObject = config.getObject(ConfigKeyName.SPARK_CONF);
        sparkConf.setAppName(config.getString(ConfigKeyName.SPARK_JOB_NAME));
        for (Map.Entry<String,Object> entry : configObject.unwrapped().entrySet()){
            if (entry.getKey().equals(ConfigKeyName.SPARK_MASTER_KEY_NAME)){
                sparkConf.setMaster(String.valueOf(entry.getValue()));
            }else {
                sparkConf.set(entry.getKey(),String.valueOf(entry.getValue()));
            }
        }
    }

    /**
     * 创建一个spark流执行环境
     * @return StreamingContext
     */
    public StreamingContext createStreamingContext(){
      SparkConf sparkConf =  sparkSession.sparkContext().getConf();
      long duration = sparkConf.getLong("spark.stream.batchDuration", config.getInt(ConfigKeyName.SPARK_BATCH_DURATION));
      streamingContext = new StreamingContext(sparkSession.sparkContext(), Seconds.apply(duration));
      return streamingContext;
    }

    /**
     * 插件获取执行环境
     * @return SparkSession
     */
    public SparkSession getSparkSession(){
        return sparkSession;
    }

    /**
     * 插件获取执行环境流执行环境
     * @return StreamingContext
     */
    public StreamingContext getStreamingContext(){
        return streamingContext;
    }

    /**
     * 返回一个临时表结构schema
     * @param jsonObject 外部定义表结构对象
     */

    public void setStructType (JSONObject jsonObject){
        schema = SchemaUtil.getTypeInformation(jsonObject);
    }

    /**
     * 用于插件获取定义的schema
     * @return  StructType
     */
    public StructType getStructType (){
       return schema;
    }

    /**
     * 用户自定义查询的临时表名称
     * @param tableName
     */
    public void setQueryTable(String tableName){
        this.tableName = tableName;
    }

    public String getQueryTable(){
        return tableName;
    }

    /**
     * 执行的SQL文件
     * @param sqlAll
     */
    public void setSqlAll(List<String> sqlAll){
        this.sqlAll = sqlAll;
    }

    public List<String> getSqlAll(){
        if (Objects.isNull(sqlAll)){
            sqlAll = new ArrayList<>();
        }
        return sqlAll;
    }
}
